/*-------------------------------------------------------------------------
 *
 * shm_mq.c
 *	  单读者，单写者共享内存消息队列
 *
 * 发送方和接收方都必须有一个 PGPROC；它们各自的进程闩用于同步。只有发送方
 * 可以发送，而只有接收方可以接收。这是为了允许用户后台与已注册的工作后台进行
 * 通信。
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * src/backend/storage/ipc/shm_mq.c
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "miscadmin.h"
#include "pgstat.h"
#include "port/pg_bitutils.h"
#include "postmaster/bgworker.h"
#include "storage/procsignal.h"
#include "storage/shm_mq.h"
#include "storage/spin.h"
#include "utils/memutils.h"


/*
 * 该结构表示实际的队列，存储在共享内存中。
 *
 * 有关同步的一些注意事项：
 *
 * mq_receiver 和 mq_bytes_read 只能由接收方更改；而
 * mq_sender 和 mq_bytes_written 只能由发送方更改。
 * mq_receiver 和 mq_sender 由 mq_mutex 保护，重要的是，
 * 一旦设定，它们就无法更改，因此在这一点已知的情况下，
 * 可以在没有锁的情况下读取。
 *
 * mq_bytes_read 和 mq_bytes_written 并未受到互斥锁的保护。相反，
 * 它们是通过 8 字节的加载和存储以原子方式写入的。必须小心使用
 * 内存屏障来同步这些值的读写与 mq_ring 中实际数据的读写。
 *
 * mq_detached 不需要锁定。它可以由发送方或接收方设置，
 * 但只能从 false 设置为 true，因此冗余写入并不
 * 重要。重要的是，如果我们设置了 mq_detached，然后设置
 * 对方的锁， 对方必须在醒来后能够确定看到该变化。
 * 由于 SetLatch 以一个内存屏障开始，而 ResetLatch
 * 以一个内存屏障结束，因此这应该没问题。
 *
 * mq_ring_size 和 mq_ring_offset 在初始化后永远不会改变，
 * 因此可以在没有锁的情况下读取。
 *
 * 重要的是，mq_ring 可以在没有锁的情况下安全地读取和写入。
 * 在任何给定时刻， mq_bytes_read 和 mq_bytes_written 之间的差
 * 量定义了 mq_ring 中包含未读数据的字节数，mq_bytes_read 定义了
 * 这些字节开始的位置。发送方可以随时增加未读字节的数量，
 * 但只有接收方可以通过递增 mq_bytes_read 来授权覆盖这些字节。
 * 因此，接收方可以在没有锁的情况下读取它知道存在的未读字节。
 * 反之，发送方可以在没有锁的情况下写入环形缓冲区的未使用部分，
 * 因为没有其他人可以读取或写入这些字节。接收方可能通过递增
 * mq_bytes_read，使更多字节未被使用，但这没问题。请注意，
 * 对于接收方来说，读取已经标记为已读的任何数据或写入任何数据
 * 是不安全的；而对于发送方来说，在递增 mq_bytes_written
 * 后重新读取任何数据也是不安全的，幸运的是没有必要进行任何
 * 这些操作。
 */ 
struct shm_mq
{
	slock_t		mq_mutex;
	PGPROC	   *mq_receiver;
	PGPROC	   *mq_sender;
	pg_atomic_uint64 mq_bytes_read;
	pg_atomic_uint64 mq_bytes_written;
	Size		mq_ring_size;
	bool		mq_detached;
	uint8		mq_ring_offset;
	char		mq_ring[FLEXIBLE_ARRAY_MEMBER];
};


/*
 * 该结构是一个后端私有句柄，用于访问队列。
 *
 * mqh_queue 是一个指向我们已连接队列的指针，mqh_segment 是一个
 * 可选的指向包含它的动态共享内存段的指针。 （如果提供 mqh_segment，
 * 我们注册一个 on_dsm_detach 回调，以确保在从 DSM 卸载之前从队列中
 * 卸载。）
 *
 * 如果该队列旨在将当前进程与启动它的后台工作者连接，用户可以将
 * 指向工作者句柄的指针传递给 shm_mq_attach()，我们会将其存储在
 * mqh_handle 中。 这样做的目的是允许我们在与之通信的进程尚未启动
 * 之前就开始向该队列发送或接收数据。如果它无法启动，句柄将使我们
 * 注意到这一点并干净地失败，而不是永远等待；请参见 shm_mq_wait_internal。
 * 这在简单情况下特别有用——例如，仅两个进程进行通信；在更复杂的场景中，
 * 每个进程可能没有可用的 BackgroundWorkerHandle，或者可能需要同时监视
 * 多个其他进程的失败。
 *
 * 当消息作为队列中连续的字节块存在时——也就是说，它小于环形缓冲区的
 * 大小并且不跨越末尾——我们将消息作为指向缓冲区的指针返回给调用者。
 * 对于较大或恰好跨越的消息，我们通过将块复制到后端本地缓冲区来
 * 在本地重新组装消息。mqh_buffer 是缓冲区，mqh_buflen 是为其分配
 * 的字节数。
 *
 * mqh_send_pending 是写入队列但尚未更新到共享内存中的字节数。
 * 在写入数据达到环形大小的四分之一或元组队列满之前，我们不会更新它。
 * 这将防止频繁的 CPU 缓存未命中，同时也避免频繁的 SetLatch() 调用，
 * 而这些调用相当昂贵。
 *
 * mqh_partial_bytes、mqh_expected_bytes 和 mqh_length_word_complete
 * 用于跟踪非阻塞操作的状态。当调用者尝试返回 SHM_MQ_WOULD_BLOCK
 * 的非阻塞操作时，他们应该在稍后使用相同的参数重试该调用；
 * 我们需要保留足够的状态以从上次中断的地方继续。
 * mqh_length_word_complete 跟踪我们是否完成了整个长度字的发送
 * 或接收（无论我们正在做什么）。mqh_partial_bytes 跟踪长度字或
 * 消息本身的读取或写入的字节数，而 mqh_expected_bytes——
 * 仅用于读取——跟踪有效负载的预期总大小。
 *
 * mqh_counterparty_attached 跟踪我们是否知道对方在之前的某个时刻
 * 已附加到队列。这使我们能够避免某些互斥锁的获取。
 *
 * mqh_context 是我们附加到 shm_mq 时生效的内存上下文。
 * shm_mq_handle 本身是在该上下文中分配的，我们确保我们进行的
 * 任何其他分配也发生在该上下文中，以避免意外情况。
 */
struct shm_mq_handle
{
	shm_mq	   *mqh_queue;
	dsm_segment *mqh_segment;
	BackgroundWorkerHandle *mqh_handle;
	char	   *mqh_buffer;
	Size		mqh_buflen;
	Size		mqh_consume_pending;
	Size		mqh_send_pending;
	Size		mqh_partial_bytes;
	Size		mqh_expected_bytes;
	bool		mqh_length_word_complete;
	bool		mqh_counterparty_attached;
	MemoryContext mqh_context;
};

static void fc_shm_mq_detach_internal(shm_mq *fc_mq);
static shm_mq_result fc_shm_mq_send_bytes(shm_mq_handle *fc_mqh, Size fc_nbytes,
									   const void *fc_data, bool fc_nowait, Size *fc_bytes_written);
static shm_mq_result fc_shm_mq_receive_bytes(shm_mq_handle *fc_mqh,
										  Size fc_bytes_needed, bool fc_nowait, Size *fc_nbytesp,
										  void **fc_datap);
static bool fc_shm_mq_counterparty_gone(shm_mq *fc_mq,
									 BackgroundWorkerHandle *fc_handle);
static bool fc_shm_mq_wait_internal(shm_mq *fc_mq, PGPROC **fc_ptr,
								 BackgroundWorkerHandle *fc_handle);
static void fc_shm_mq_inc_bytes_read(shm_mq *fc_mq, Size fc_n);
static void fc_shm_mq_inc_bytes_written(shm_mq *fc_mq, Size fc_n);
static void fc_shm_mq_detach_callback(dsm_segment *fc_seg, Datum fc_arg);

/* 最小队列大小足以容纳头部和至少一块数据。 */
const Size	shm_mq_minimum_size =
MAXALIGN(offsetof(shm_mq, mq_ring)) + MAXIMUM_ALIGNOF;

#define MQH_INITIAL_BUFSIZE				8192

/*
 * 初始化一个新的共享消息队列。
 */
shm_mq *
shm_mq_create(void *fc_address, Size fc_size)
{
	shm_mq	   *fc_mq = fc_address;
	Size		fc_data_offset = MAXALIGN(offsetof(shm_mq, mq_ring));

	/* 如果大小不是 MAXALIGN 的，则丢弃多余的字节。 */
	fc_size = MAXALIGN_DOWN(fc_size);

	/* 队列大小必须足够大以容纳一些数据。 */
	Assert(fc_size > fc_data_offset);

	/* 初始化队列头。 */
	SpinLockInit(&fc_mq->mq_mutex);
	fc_mq->mq_receiver = NULL;
	fc_mq->mq_sender = NULL;
	pg_atomic_init_u64(&fc_mq->mq_bytes_read, 0);
	pg_atomic_init_u64(&fc_mq->mq_bytes_written, 0);
	fc_mq->mq_ring_size = fc_size - fc_data_offset;
	fc_mq->mq_detached = false;
	fc_mq->mq_ring_offset = fc_data_offset - offsetof(shm_mq, mq_ring);

	return fc_mq;
}

/*
 * 设置将从共享消息队列接收的进程的身份。
 */
void shm_mq_set_receiver(shm_mq *fc_mq, PGPROC *fc_proc)
{
	PGPROC	   *fc_sender;

	SpinLockAcquire(&fc_mq->mq_mutex);
	Assert(fc_mq->mq_receiver == NULL);
	fc_mq->mq_receiver = fc_proc;
	fc_sender = fc_mq->mq_sender;
	SpinLockRelease(&fc_mq->mq_mutex);

	if (fc_sender != NULL)
		SetLatch(&fc_sender->procLatch);
}

/*
 * 设置将向共享消息队列发送的进程的身份。
 */
void shm_mq_set_sender(shm_mq *fc_mq, PGPROC *fc_proc)
{
	PGPROC	   *fc_receiver;

	SpinLockAcquire(&fc_mq->mq_mutex);
	Assert(fc_mq->mq_sender == NULL);
	fc_mq->mq_sender = fc_proc;
	fc_receiver = fc_mq->mq_receiver;
	SpinLockRelease(&fc_mq->mq_mutex);

	if (fc_receiver != NULL)
		SetLatch(&fc_receiver->procLatch);
}

/*
 * 获取配置的接收者。
 */
PGPROC * shm_mq_get_receiver(shm_mq *fc_mq)
{
	PGPROC	   *fc_receiver;

	SpinLockAcquire(&fc_mq->mq_mutex);
	fc_receiver = fc_mq->mq_receiver;
	SpinLockRelease(&fc_mq->mq_mutex);

	return fc_receiver;
}

/*
 * 获取配置的发送者。
 */
PGPROC * shm_mq_get_sender(shm_mq *fc_mq)
{
	PGPROC	   *fc_sender;

	SpinLockAcquire(&fc_mq->mq_mutex);
	fc_sender = fc_mq->mq_sender;
	SpinLockRelease(&fc_mq->mq_mutex);

	return fc_sender;
}

/*
 * 附加到共享消息队列，以便我们可以发送或接收消息。
 *
 * 当调用此函数时生效的内存上下文应至少持续到消息队列本身存活的
 * 时间。我们将在该上下文中分配句柄，并且后续需要缓冲传入数据的
 * 分配也将在该上下文中发生。
 *
 * 如果 seg != NULL，当该动态共享内存段被卸载时，队列将自动被
 * 卸载。
 *
 * 如果 handle != NULL，即使其他进程尚未附加，队列也可以被读取或
 * 写入。我们会在需要时等待它这样做。该句柄必须对应于一个背景工作者，
 * 其 bgw_notify_pid 等于我们的 PID。
 *
 * 完成后应调用 shm_mq_detach()。这将释放 shm_mq_handle，并将
 * 队列本身标记为已卸载，以便我们的对方不会在我们已经失去兴趣后
 * 被卡在等待我们填充或排空队列中。
 */
shm_mq_handle *
shm_mq_attach(shm_mq *fc_mq, dsm_segment *fc_seg, BackgroundWorkerHandle *fc_handle)
{
	shm_mq_handle *fc_mqh = palloc(sizeof(shm_mq_handle));

	Assert(fc_mq->mq_receiver == MyProc || fc_mq->mq_sender == MyProc);
	fc_mqh->mqh_queue = fc_mq;
	fc_mqh->mqh_segment = fc_seg;
	fc_mqh->mqh_handle = fc_handle;
	fc_mqh->mqh_buffer = NULL;
	fc_mqh->mqh_buflen = 0;
	fc_mqh->mqh_consume_pending = 0;
	fc_mqh->mqh_send_pending = 0;
	fc_mqh->mqh_partial_bytes = 0;
	fc_mqh->mqh_expected_bytes = 0;
	fc_mqh->mqh_length_word_complete = false;
	fc_mqh->mqh_counterparty_attached = false;
	fc_mqh->mqh_context = CurrentMemoryContext;

	if (fc_seg != NULL)
		on_dsm_detach(fc_seg, fc_shm_mq_detach_callback, PointerGetDatum(fc_mq));

	return fc_mqh;
}

/*
 * 将 BackgroundWorkerHandle 与 shm_mq_handle 关联，就如同将其
 * 传递给 shm_mq_attach 一样。
 */
void shm_mq_set_handle(shm_mq_handle *fc_mqh, BackgroundWorkerHandle *fc_handle)
{
	Assert(fc_mqh->mqh_handle == NULL);
	fc_mqh->mqh_handle = fc_handle;
}

/*
 * 将消息写入共享消息队列。
 */
shm_mq_result shm_mq_send(shm_mq_handle *fc_mqh, Size fc_nbytes, const void *fc_data, bool fc_nowait,
			bool fc_force_flush)
{
	shm_mq_iovec fc_iov;

	fc_iov.data = fc_data;
	fc_iov.len = fc_nbytes;

	return shm_mq_sendv(fc_mqh, &fc_iov, 1, fc_nowait, fc_force_flush);
}


shm_mq_result shm_mq_sendv(shm_mq_handle *fc_mqh, shm_mq_iovec *fc_iov, int fc_iovcnt, bool fc_nowait,
			 bool fc_force_flush)
{
	shm_mq_result fc_res;
	shm_mq	   *fc_mq = fc_mqh->mqh_queue;
	PGPROC	   *fc_receiver;
	Size		fc_nbytes = 0;
	Size		fc_bytes_written;
	int			fc_i;
	int			fc_which_iov = 0;
	Size		fc_offset;

	Assert(fc_mq->mq_sender == MyProc);

	/* 计算写入的总大小。 */
	for (fc_i = 0; fc_i < fc_iovcnt; ++fc_i)
		fc_nbytes += fc_iov[fc_i].len;

	/* 防止写入消息压倒接收者。 */
	if (fc_nbytes > MaxAllocSize)
		ereport(ERROR,
				(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
				 errmsg("cannot send a message of size %zu via shared memory queue",
						fc_nbytes)));

	/* 尝试将长度字写入缓冲区，或完成写入。 */
	while (!fc_mqh->mqh_length_word_complete)
	{
		Assert(fc_mqh->mqh_partial_bytes < sizeof(Size));
		fc_res = fc_shm_mq_send_bytes(fc_mqh, sizeof(Size) - fc_mqh->mqh_partial_bytes,
								((char *) &fc_nbytes) + fc_mqh->mqh_partial_bytes,
								fc_nowait, &fc_bytes_written);

		if (fc_res == SHM_MQ_DETACHED)
		{
			/* 重置状态，以防调用方尝试发送另一条消息。 */
			fc_mqh->mqh_partial_bytes = 0;
			fc_mqh->mqh_length_word_complete = false;
			return fc_res;
		}
		fc_mqh->mqh_partial_bytes += fc_bytes_written;

		if (fc_mqh->mqh_partial_bytes >= sizeof(Size))
		{
			Assert(fc_mqh->mqh_partial_bytes == sizeof(Size));

			fc_mqh->mqh_partial_bytes = 0;
			fc_mqh->mqh_length_word_complete = true;
		}

		if (fc_res != SHM_MQ_SUCCESS)
			return fc_res;

		/* 除非大于所需对齐，长度字不能被拆分。 */
		Assert(fc_mqh->mqh_length_word_complete || sizeof(Size) > MAXIMUM_ALIGNOF);
	}

	/* 将实际数据字节写入缓冲区。 */
	Assert(fc_mqh->mqh_partial_bytes <= fc_nbytes);
	fc_offset = fc_mqh->mqh_partial_bytes;
	do
	{
		Size		fc_chunksize;

		/* 找出需要发送的字节。 */
		if (fc_offset >= fc_iov[fc_which_iov].len)
		{
			fc_offset -= fc_iov[fc_which_iov].len;
			++fc_which_iov;
			if (fc_which_iov >= fc_iovcnt)
				break;
			continue;
		}

		/*
		 * 我们希望尽可能避免复制数据，但我们写入队列的每个
		 * 字节块都必须进行对齐，除了最后一个。因此，如果
		 * 除了最后一个块外的某个块以非对齐的边界结束，我们必须
		 * 将其数据的尾部与一个或多个后续块的数据组合，直到我们
		 * 到达最后一个块或累积出一个对齐的字节数。
		 */
		if (fc_which_iov + 1 < fc_iovcnt &&
			fc_offset + MAXIMUM_ALIGNOF > fc_iov[fc_which_iov].len)
		{
			char		fc_tmpbuf[MAXIMUM_ALIGNOF];
			int			fc_j = 0;

			for (;;)
			{
				if (fc_offset < fc_iov[fc_which_iov].len)
				{
					fc_tmpbuf[fc_j] = fc_iov[fc_which_iov].data[fc_offset];
					fc_j++;
					fc_offset++;
					if (fc_j == MAXIMUM_ALIGNOF)
						break;
				}
				else
				{
					fc_offset -= fc_iov[fc_which_iov].len;
					fc_which_iov++;
					if (fc_which_iov >= fc_iovcnt)
						break;
				}
			}

			fc_res = fc_shm_mq_send_bytes(fc_mqh, fc_j, fc_tmpbuf, fc_nowait, &fc_bytes_written);

			if (fc_res == SHM_MQ_DETACHED)
			{
				/* 重置状态，以防调用方尝试发送另一条消息。 */
				fc_mqh->mqh_partial_bytes = 0;
				fc_mqh->mqh_length_word_complete = false;
				return fc_res;
			}

			fc_mqh->mqh_partial_bytes += fc_bytes_written;
			if (fc_res != SHM_MQ_SUCCESS)
				return fc_res;
			continue;
		}

		/*
		 * 如果这是最后一个块，我们可以写入所有数据，即使它
		 * 不是最大对齐的倍数。否则，我们需要将写入大小下
		 * 限对齐。
		 */
		fc_chunksize = fc_iov[fc_which_iov].len - fc_offset;
		if (fc_which_iov + 1 < fc_iovcnt)
			fc_chunksize = MAXALIGN_DOWN(fc_chunksize);
		fc_res = fc_shm_mq_send_bytes(fc_mqh, fc_chunksize, &fc_iov[fc_which_iov].data[fc_offset],
								fc_nowait, &fc_bytes_written);

		if (fc_res == SHM_MQ_DETACHED)
		{
			/* 重置状态，以防调用方尝试发送另一条消息。 */
			fc_mqh->mqh_length_word_complete = false;
			fc_mqh->mqh_partial_bytes = 0;
			return fc_res;
		}

		fc_mqh->mqh_partial_bytes += fc_bytes_written;
		fc_offset += fc_bytes_written;
		if (fc_res != SHM_MQ_SUCCESS)
			return fc_res;
	} while (fc_mqh->mqh_partial_bytes < fc_nbytes);

	/* 为下一条消息重置。 */
	fc_mqh->mqh_partial_bytes = 0;
	fc_mqh->mqh_length_word_complete = false;

	/* 如果队列已被分离，通知调用方。 */
	if (fc_mq->mq_detached)
		return SHM_MQ_DETACHED;

	/*
	 * 如果已知对方已连接，我们可以在不获取自旋锁的情况下
	 * 读取 mq_receiver。否则，需要更加小心。
	 */
	if (fc_mqh->mqh_counterparty_attached)
		fc_receiver = fc_mq->mq_receiver;
	else
	{
		SpinLockAcquire(&fc_mq->mq_mutex);
		fc_receiver = fc_mq->mq_receiver;
		SpinLockRelease(&fc_mq->mq_mutex);
		if (fc_receiver != NULL)
			fc_mqh->mqh_counterparty_attached = true;
	}

	/*
	 * 如果调用方请求强制刷新或我们已经写入超过 1/4 的
	 * 环大小，则在共享内存中标记为已写入并通知接收者。
	 */
	if (fc_force_flush || fc_mqh->mqh_send_pending > (fc_mq->mq_ring_size >> 2))
	{
		fc_shm_mq_inc_bytes_written(fc_mq, fc_mqh->mqh_send_pending);
		if (fc_receiver != NULL)
			SetLatch(&fc_receiver->procLatch);
		fc_mqh->mqh_send_pending = 0;
	}

	return SHM_MQ_SUCCESS;
}

/*
 * 从共享消息队列接收消息。
 *
 * 我们将 *nbytes 设置为消息长度，并将 *data 指向消息
 * 负载。如果整个消息作为一个单一、连续的块存在于队列中，
 * *data 将直接指向共享内存；否则，它将指向临时缓冲区。
 * 这样大部分避免了数据复制，假设消息相对于缓冲区大小较短，
 * 同时仍然允许较长的消息。在任一情况下，返回值在下次
 * 对队列执行接收操作之前保持有效。
 *
 * 当 nowait = false 时，当环缓冲区为空且我们尚未接收到
 * 完整消息时，我们将等待我们的进程闩。当更多数据被
 * 写入时，发送方将设置我们的进程闩，我们将恢复处理。
 * 因此，每次调用将返回完整消息（除非发送方分离了队列）。
 *
 * 当 nowait = true 时，我们不操作进程闩的状态；
 * 相反，每当缓冲区为空且我们需要从中读取时，我们
 * 返回 SHM_MQ_WOULD_BLOCK。在这种情况下，调用方应
 * 在进程闩设置后再次调用此函数。
 */
shm_mq_result shm_mq_receive(shm_mq_handle *fc_mqh, Size *fc_nbytesp, void **fc_datap, bool fc_nowait)
{
	shm_mq	   *fc_mq = fc_mqh->mqh_queue;
	shm_mq_result fc_res;
	Size		fc_rb = 0;
	Size		fc_nbytes;
	void	   *fc_rawdata;

	Assert(fc_mq->mq_receiver == MyProc);

	/* 在发送方未连接之前，我们无法接收数据。 */
	if (!fc_mqh->mqh_counterparty_attached)
	{
		if (fc_nowait)
		{
			int			fc_counterparty_gone;

			/*
			 * 在这一点上，我们不应该返回，除非发送方还没有
			 * 连接。然而，正确的返回值取决于发送方是否仍然
			 * 连接。如果我们首先检查发送方是否曾连接，然后
			 * 检查发送方是否已分离，就会出现竞争条件：一个
			 * 迅速连接和分离的发送方可能会误导我们认为发送方
			 * 根本没有连接。因此，先检查我们的对方是否已
			 * 确定消失，然后再检查发送方是否曾连接。
			 */
			fc_counterparty_gone = fc_shm_mq_counterparty_gone(fc_mq, fc_mqh->mqh_handle);
			if (shm_mq_get_sender(fc_mq) == NULL)
			{
				if (fc_counterparty_gone)
					return SHM_MQ_DETACHED;
				else
					return SHM_MQ_WOULD_BLOCK;
			}
		}
		else if (!fc_shm_mq_wait_internal(fc_mq, &fc_mq->mq_sender, fc_mqh->mqh_handle)
				 && shm_mq_get_sender(fc_mq) == NULL)
		{
			fc_mq->mq_detached = true;
			return SHM_MQ_DETACHED;
		}
		fc_mqh->mqh_counterparty_attached = true;
	}

	/*
	 * 如果我们消耗的数据量超过环大小的 1/4，则在共享内存
	 * 中标记为已消耗。我们尽量避免在仅消耗少量数据时不必要
	 * 地进行此操作，因为 SetLatch() 相对昂贵，我们不想过
	 * 于频繁地进行此操作。
	 */
	if (fc_mqh->mqh_consume_pending > fc_mq->mq_ring_size / 4)
	{
		fc_shm_mq_inc_bytes_read(fc_mq, fc_mqh->mqh_consume_pending);
		fc_mqh->mqh_consume_pending = 0;
	}

	/* 尝试从缓冲区读取长度字，或完成读取。 */
	while (!fc_mqh->mqh_length_word_complete)
	{
		/* 尝试接收消息长度字。 */
		Assert(fc_mqh->mqh_partial_bytes < sizeof(Size));
		fc_res = fc_shm_mq_receive_bytes(fc_mqh, sizeof(Size) - fc_mqh->mqh_partial_bytes,
								   fc_nowait, &fc_rb, &fc_rawdata);
		if (fc_res != SHM_MQ_SUCCESS)
			return fc_res;

		/*
		 * 希望我们可以一次接收到整个消息长度字。
		 * 但如果 sizeof(Size) > MAXIMUM_ALIGNOF，那么它可能会
		 * 在多次读取中被拆分。
		 */
		if (fc_mqh->mqh_partial_bytes == 0 && fc_rb >= sizeof(Size))
		{
			Size		fc_needed;

			fc_nbytes = *(Size *) fc_rawdata;

			/* 如果我们已经获得整个消息，我们就结束了。 */
			fc_needed = MAXALIGN(sizeof(Size)) + MAXALIGN(fc_nbytes);
			if (fc_rb >= fc_needed)
			{
				fc_mqh->mqh_consume_pending += fc_needed;
				*fc_nbytesp = fc_nbytes;
				*fc_datap = ((char *) fc_rawdata) + MAXALIGN(sizeof(Size));
				return SHM_MQ_SUCCESS;
			}

			/*
			 * 我们没有整个消息，但至少我们有整个长度字。
			 */
			fc_mqh->mqh_expected_bytes = fc_nbytes;
			fc_mqh->mqh_length_word_complete = true;
			fc_mqh->mqh_consume_pending += MAXALIGN(sizeof(Size));
			fc_rb -= MAXALIGN(sizeof(Size));
		}
		else
		{
			Size		fc_lengthbytes;

			/* 除非大于所需对齐，否则无法拆分。 */
			Assert(sizeof(Size) > MAXIMUM_ALIGNOF);

			/* 消息字被拆分；需要缓冲区进行重新组装。 */
			if (fc_mqh->mqh_buffer == NULL)
			{
				fc_mqh->mqh_buffer = MemoryContextAlloc(fc_mqh->mqh_context,
													 MQH_INITIAL_BUFSIZE);
				fc_mqh->mqh_buflen = MQH_INITIAL_BUFSIZE;
			}
			Assert(fc_mqh->mqh_buflen >= sizeof(Size));

			/* 复制部分长度字；记得消耗它。 */
			if (fc_mqh->mqh_partial_bytes + fc_rb > sizeof(Size))
				fc_lengthbytes = sizeof(Size) - fc_mqh->mqh_partial_bytes;
			else
				fc_lengthbytes = fc_rb;
			memcpy(&fc_mqh->mqh_buffer[fc_mqh->mqh_partial_bytes], fc_rawdata,
				   fc_lengthbytes);
			fc_mqh->mqh_partial_bytes += fc_lengthbytes;
			fc_mqh->mqh_consume_pending += MAXALIGN(fc_lengthbytes);
			fc_rb -= fc_lengthbytes;

			/* 如果我们现在有整个字，我们准备读取负载。 */
			if (fc_mqh->mqh_partial_bytes >= sizeof(Size))
			{
				Assert(fc_mqh->mqh_partial_bytes == sizeof(Size));
				fc_mqh->mqh_expected_bytes = *(Size *) fc_mqh->mqh_buffer;
				fc_mqh->mqh_length_word_complete = true;
				fc_mqh->mqh_partial_bytes = 0;
			}
		}
	}
	fc_nbytes = fc_mqh->mqh_expected_bytes;

	/*
	 * 在发送端就应该禁止，但在接收端最好检查并报错，
	 * 而不是尝试读取一个过于大的消息。
	 */
	if (fc_nbytes > MaxAllocSize)
		ereport(ERROR,
				(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
				 errmsg("invalid message size %zu in shared memory queue",
						fc_nbytes)));

	if (fc_mqh->mqh_partial_bytes == 0)
	{
		/*
		 * 尝试在一个块中获取整个消息。如果成功，我们就不需要
		 * 复制数据，可以直接返回指向共享内存的指针。
		 */
		fc_res = fc_shm_mq_receive_bytes(fc_mqh, fc_nbytes, fc_nowait, &fc_rb, &fc_rawdata);
		if (fc_res != SHM_MQ_SUCCESS)
			return fc_res;
		if (fc_rb >= fc_nbytes)
		{
			fc_mqh->mqh_length_word_complete = false;
			fc_mqh->mqh_consume_pending += MAXALIGN(fc_nbytes);
			*fc_nbytesp = fc_nbytes;
			*fc_datap = fc_rawdata;
			return SHM_MQ_SUCCESS;
		}

		
		if (fc_mqh->mqh_buflen < fc_nbytes)
		{
			Size		fc_newbuflen;

			/*
			 * 增加大小到大于等于 nbytes 的下一个 2 的幂，但
			 * 限制在 MaxAllocSize 内。
			 */
			fc_newbuflen = pg_nextpower2_size_t(fc_nbytes);
			fc_newbuflen = Min(fc_newbuflen, MaxAllocSize);

			if (fc_mqh->mqh_buffer != NULL)
			{
				pfree(fc_mqh->mqh_buffer);
				fc_mqh->mqh_buffer = NULL;
				fc_mqh->mqh_buflen = 0;
			}
			fc_mqh->mqh_buffer = MemoryContextAlloc(fc_mqh->mqh_context, fc_newbuflen);
			fc_mqh->mqh_buflen = fc_newbuflen;
		}
	}

	/* 循环直到我们复制完整个消息。 */
	for (;;)
	{
		Size		fc_still_needed;

		/* 尽可能多地复制。 */
		Assert(fc_mqh->mqh_partial_bytes + fc_rb <= fc_nbytes);
		if (fc_rb > 0)
		{
			memcpy(&fc_mqh->mqh_buffer[fc_mqh->mqh_partial_bytes], fc_rawdata, fc_rb);
			fc_mqh->mqh_partial_bytes += fc_rb;
		}

		/*
		 * 更新可消费字节的计数，考虑到对齐填充。注意，这实际上
		 * 不会插入任何填充，除非在消息的末尾，因为缓冲区大小是
		 * MAXIMUM_ALIGNOF 的倍数，并且每次读取和写入也是如此。
		 */
		Assert(fc_mqh->mqh_partial_bytes == fc_nbytes || fc_rb == MAXALIGN(fc_rb));
		fc_mqh->mqh_consume_pending += MAXALIGN(fc_rb);

		/* 如果我们已经得到了所有数据，退出循环。 */
		if (fc_mqh->mqh_partial_bytes >= fc_nbytes)
			break;

		/* 等待更多数据。 */
		fc_still_needed = fc_nbytes - fc_mqh->mqh_partial_bytes;
		fc_res = fc_shm_mq_receive_bytes(fc_mqh, fc_still_needed, fc_nowait, &fc_rb, &fc_rawdata);
		if (fc_res != SHM_MQ_SUCCESS)
			return fc_res;
		if (fc_rb > fc_still_needed)
			fc_rb = fc_still_needed;
	}

	/* 返回完整消息，并重置为下一个消息。 */
	*fc_nbytesp = fc_nbytes;
	*fc_datap = fc_mqh->mqh_buffer;
	fc_mqh->mqh_length_word_complete = false;
	fc_mqh->mqh_partial_bytes = 0;
	return SHM_MQ_SUCCESS;
}

/*
 * 等待另一个应该使用此队列的进程附加到它。
 *
 * 如果工作线程已经分离或死掉，返回值为 SHM_MQ_DETACHED；如果
 * 检测到工作线程已附加，则为 SHM_MQ_SUCCESS。注意，只有在
 * 附加之前，我们才能检测到工作线程是否已经死亡，前提是
 * 有一个后台工作线程句柄被传递给 shm_mq_attach()。
 */
shm_mq_result shm_mq_wait_for_attach(shm_mq_handle *fc_mqh)
{
	shm_mq	   *fc_mq = fc_mqh->mqh_queue;
	PGPROC	  **fc_victim;

	if (shm_mq_get_receiver(fc_mq) == MyProc)
		fc_victim = &fc_mq->mq_sender;
	else
	{
		Assert(shm_mq_get_sender(fc_mq) == MyProc);
		fc_victim = &fc_mq->mq_receiver;
	}

	if (fc_shm_mq_wait_internal(fc_mq, fc_victim, fc_mqh->mqh_handle))
		return SHM_MQ_SUCCESS;
	else
		return SHM_MQ_DETACHED;
}

/*
 * 从共享消息队列分离，并销毁 shm_mq_handle。
 */
void shm_mq_detach(shm_mq_handle *fc_mqh)
{
	/* 在分离之前，通知接收方有关已写入的数据。 */
	if (fc_mqh->mqh_send_pending > 0)
	{
		fc_shm_mq_inc_bytes_written(fc_mqh->mqh_queue, fc_mqh->mqh_send_pending);
		fc_mqh->mqh_send_pending = 0;
	}

	/* 通知对方我们要离开了。 */
	fc_shm_mq_detach_internal(fc_mqh->mqh_queue);

	/* 取消 on_dsm_detach 回调（如果有）。 */
	if (fc_mqh->mqh_segment)
		cancel_on_dsm_detach(fc_mqh->mqh_segment,
							 fc_shm_mq_detach_callback,
							 PointerGetDatum(fc_mqh->mqh_queue));

	/* 释放与句柄相关的本地内存。 */
	if (fc_mqh->mqh_buffer != NULL)
		pfree(fc_mqh->mqh_buffer);
	pfree(fc_mqh);
}

/*
 * 通知对方我们正在从共享消息队列分离。
 *
 * 此函数的目的是确保我们通信的进程不会因为我们
 * 失去兴趣而永远阻塞，等待我们填满或 draining 队列。
 * 当发送者分离时，接收方可以读取队列中剩余的任何消息；
 * 进一步的读取将返回 SHM_MQ_DETACHED。如果接收方分离，
 * 进一步尝试发送消息也将返回 SHM_MQ_DETACHED。
 *
 * 这与 shm_mq_detach() 分开，因为如果 on_dsm_detach
 * 回调被触发，我们只想做这么多。我们不试图触碰
 * 本地 shm_mq_handle，因为它可能已经被 pfree 了。
 */
static void fc_shm_mq_detach_internal(shm_mq *fc_mq)
{
	PGPROC	   *fc_victim;

	SpinLockAcquire(&fc_mq->mq_mutex);
	if (fc_mq->mq_sender == MyProc)
		fc_victim = fc_mq->mq_receiver;
	else
	{
		Assert(fc_mq->mq_receiver == MyProc);
		fc_victim = fc_mq->mq_sender;
	}
	fc_mq->mq_detached = true;
	SpinLockRelease(&fc_mq->mq_mutex);

	if (fc_victim != NULL)
		SetLatch(&fc_victim->procLatch);
}

/*
 * 从句柄中获取 shm_mq。
 */
shm_mq *
shm_mq_get_queue(shm_mq_handle *fc_mqh)
{
	return fc_mqh->mqh_queue;
}

/*
 * 将字节写入共享消息队列。
 */
static shm_mq_result
fc_shm_mq_send_bytes(shm_mq_handle *fc_mqh, Size fc_nbytes, const void *fc_data,
				  bool fc_nowait, Size *fc_bytes_written)
{
	shm_mq	   *fc_mq = fc_mqh->mqh_queue;
	Size		fc_sent = 0;
	uint64		fc_used;
	Size		fc_ringsize = fc_mq->mq_ring_size;
	Size		fc_available;

	while (fc_sent < fc_nbytes)
	{
		uint64		fc_rb;
		uint64		fc_wb;

		/* 计算环形缓冲区已用和可用字节数。 */
		fc_rb = pg_atomic_read_u64(&fc_mq->mq_bytes_read);
		fc_wb = pg_atomic_read_u64(&fc_mq->mq_bytes_written) + fc_mqh->mqh_send_pending;
		Assert(fc_wb >= fc_rb);
		fc_used = fc_wb - fc_rb;
		Assert(fc_used <= fc_ringsize);
		fc_available = Min(fc_ringsize - fc_used, fc_nbytes - fc_sent);

		/*
		 * 如果队列已被分离则退出。注意，如果编译器决
		 * 定缓存 mq->mq_detached 的值到寄存器或栈
		 * 中跨越循环迭代，我们将会遇到麻烦。它可能
		 * 不应该这样做，因为我们总会返回、调用执行
		 * 系统调用的外部函数，或者在循环的某个点
		 * 达到内存屏障，但为了安全起见，在这里插入
		 * 一个编译器屏障。
		 */
		pg_compiler_barrier();
		if (fc_mq->mq_detached)
		{
			*fc_bytes_written = fc_sent;
			return SHM_MQ_DETACHED;
		}

		if (fc_available == 0 && !fc_mqh->mqh_counterparty_attached)
		{
			/*
			 * 队列已满，因此如果接收方尚未附加，
			 * 我们必须等待发生这些情况。
			 */
			if (fc_nowait)
			{
				if (fc_shm_mq_counterparty_gone(fc_mq, fc_mqh->mqh_handle))
				{
					*fc_bytes_written = fc_sent;
					return SHM_MQ_DETACHED;
				}
				if (shm_mq_get_receiver(fc_mq) == NULL)
				{
					*fc_bytes_written = fc_sent;
					return SHM_MQ_WOULD_BLOCK;
				}
			}
			else if (!fc_shm_mq_wait_internal(fc_mq, &fc_mq->mq_receiver,
										   fc_mqh->mqh_handle))
			{
				fc_mq->mq_detached = true;
				*fc_bytes_written = fc_sent;
				return SHM_MQ_DETACHED;
			}
			fc_mqh->mqh_counterparty_attached = true;

			/*
			 * 接收方在附加后可能已读取一些数据，因此
			 * 我们不得不在不重新检查队列状态的情况下
			 * 等待。
			 */
		}
		else if (fc_available == 0)
		{
			/* 更新共享内存中的待发送字节。 */
			fc_shm_mq_inc_bytes_written(fc_mq, fc_mqh->mqh_send_pending);

			/*
			 * 由于此时已知 mq->mqh_counterparty_attached 为真，
			 * mq_receiver 已被设置，一旦设置就无法更改。
			 * 因此，我们可以在不获取自旋锁的情况下读取。
			 */
			Assert(fc_mqh->mqh_counterparty_attached);
			SetLatch(&fc_mq->mq_receiver->procLatch);

			/*
			 * 我们刚刚更新了共享内存中的 mqh_send_pending 字节
			 * ，所以重置它。
			 */
			fc_mqh->mqh_send_pending = 0;

			/* 如果 nowait = true，则跳过对我们的锁的操作。 */
			if (fc_nowait)
			{
				*fc_bytes_written = fc_sent;
				return SHM_MQ_WOULD_BLOCK;
			}

			/*
			 * 等待我们的锁被设置。它可能已经因某些
			 * 不相关的原因被设置，但这仅会导致
			 * 在循环中额外进行一次遍历。为了避免在
			 * 循环顶端重置锁，值得这么做，因为设置已
			 * 设置的锁比设置一个已重置的锁要便宜得多。
			 */
			(void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
							 WAIT_EVENT_MQ_SEND);

			/* 重置锁以避免自旋。 */
			ResetLatch(MyLatch);

			/* 在我们等待时可能发生了中断。 */
			CHECK_FOR_INTERRUPTS();
		}
		else
		{
			Size		fc_offset;
			Size		fc_sendnow;

			fc_offset = fc_wb % (uint64) fc_ringsize;
			fc_sendnow = Min(fc_available, fc_ringsize - fc_offset);

			/*
			 * 尽可能多地通过单个 memcpy() 写入数据。确保
			 * 这些写入发生在上面读取 mq_bytes_read 之后。
			 * 此屏障与 shm_mq_inc_bytes_read 中的屏障配对。
			 * （由于我们将 mq_bytes_read 的读取与后续写入 mq_ring
			 * 分开，因此我们需要在这里设置完整屏障。）
			 */
			pg_memory_barrier();
			memcpy(&fc_mq->mq_ring[fc_mq->mq_ring_offset + fc_offset],
				   (char *) fc_data + fc_sent, fc_sendnow);
			fc_sent += fc_sendnow;

			/*
			 * 更新已写入字节的计数，带有对齐填充。注意
			 * 这实际上不会在一串字节的末尾插入任何填充，
			 * 因为缓冲区大小是 MAXIMUM_ALIGNOF 的倍数，
			 * 并且每次读取也是如此。
			 */
			Assert(fc_sent == fc_nbytes || fc_sendnow == MAXALIGN(fc_sendnow));

			/*
			 * 为了效率，我们不更新共享内存中的已写字节，
			 * 也不在这里设置读取器的锁。请参考
			 * shm_mq_handle 结构顶部的注释以获取更多信息。
			 */
			fc_mqh->mqh_send_pending += MAXALIGN(fc_sendnow);
		}
	}

	*fc_bytes_written = fc_sent;
	return SHM_MQ_SUCCESS;
}

/*
 * 等待至少 *nbytesp 字节可以从共享消息队列中读取，或者直到缓冲区环绕。如果队列已被分离，则返回 SHM_MQ_DETACHED。如果指定了 nowait 且需要等待，则返回 SHM_MQ_WOULD_BLOCK。否则，*datap 被设置为可以读取数据字节的位置，*nbytesp 被设置为在该地址可以读取的字节数，返回值为 SHM_MQ_SUCCESS。
 */
static shm_mq_result
fc_shm_mq_receive_bytes(shm_mq_handle *fc_mqh, Size fc_bytes_needed, bool fc_nowait,
					 Size *fc_nbytesp, void **fc_datap)
{
	shm_mq	   *fc_mq = fc_mqh->mqh_queue;
	Size		fc_ringsize = fc_mq->mq_ring_size;
	uint64		fc_used;
	uint64		fc_written;

	for (;;)
	{
		Size		fc_offset;
		uint64		fc_read;

		/* 获取已写入的字节，以便我们可以计算可读取的字节数。 */
		fc_written = pg_atomic_read_u64(&fc_mq->mq_bytes_written);

		/*
		 * 获取已读取的字节。包括我们可以消费但尚未消耗的字节。
		 */
		fc_read = pg_atomic_read_u64(&fc_mq->mq_bytes_read) +
			fc_mqh->mqh_consume_pending;
		fc_used = fc_written - fc_read;
		Assert(fc_used <= fc_ringsize);
		fc_offset = fc_read % (uint64) fc_ringsize;

		/* 如果我们拥有足够的数据或缓冲区已环绕，我们就完成了。 */
		if (fc_used >= fc_bytes_needed || fc_offset + fc_used >= fc_ringsize)
		{
			*fc_nbytesp = Min(fc_used, fc_ringsize - fc_offset);
			*fc_datap = &fc_mq->mq_ring[fc_mq->mq_ring_offset + fc_offset];

			/*
			 * 将 mq_bytes_written 的读取与调用者尝试读取数据本身分开。与 shm_mq_inc_bytes_written 中的屏障配对。
			 */
			pg_read_barrier();
			return SHM_MQ_SUCCESS;
		}

		/*
		 * 如果队列已被分离，则在等待之前退出。
		 *
		 * 请注意，我们在*考虑现有可用数据是否足够*后，才检查这一点，因为接收者可以在发送者已分离后完成接收存储在缓冲区中的消息。
		 */
		if (fc_mq->mq_detached)
		{
			/*
			 * 如果写入者提前更新了 mq_bytes_written 然后设置了 mq_detached，我们可能没有读取 mq_bytes_written 的最终值。插入一个读取屏障，然后再次检查 mq_bytes_written 是否已增加。
			 */
			pg_read_barrier();
			if (fc_written != pg_atomic_read_u64(&fc_mq->mq_bytes_written))
				continue;

			return SHM_MQ_DETACHED;
		}

		/*
		 * 我们没有获得足够的数据来满足请求，因此将先前消费的数据标记为已读取，以释放更多的缓冲区空间。
		 */
		if (fc_mqh->mqh_consume_pending > 0)
		{
			fc_shm_mq_inc_bytes_read(fc_mq, fc_mqh->mqh_consume_pending);
			fc_mqh->mqh_consume_pending = 0;
		}

		/* 如果 nowait = true，则跳过对我们的锁的操作。 */
		if (fc_nowait)
			return SHM_MQ_WOULD_BLOCK;

		/*
		 * 等待我们的锁被设置。它可能已经因为某些无关的原因被设置，但这只会导致在循环中多走一次。避免在循环开头重置锁是值得的，因为设置一个已经设置的锁要比设置一个已重置的锁便宜得多。
		 */
		(void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
						 WAIT_EVENT_MQ_RECEIVE);

		/* 重置锁以避免自旋。 */
		ResetLatch(MyLatch);

		/* 在我们等待时可能发生了中断。 */
		CHECK_FOR_INTERRUPTS();
	}
}

/*
 * 测试一个可能甚至还没有活着的对方是否 definitivamente 已无。
 */
static bool fc_shm_mq_counterparty_gone(shm_mq *fc_mq, BackgroundWorkerHandle *fc_handle)
{
	pid_t		fc_pid;

	/* 如果队列已被分离，则对方肯定已经消失。 */
	if (fc_mq->mq_detached)
		return true;

	/* 如果有句柄，检查工作者状态。 */
	if (fc_handle != NULL)
	{
		BgwHandleStatus fc_status;

		/* 检查意外的工作者死亡。 */
		fc_status = GetBackgroundWorkerPid(fc_handle, &fc_pid);
		if (fc_status != BGWH_STARTED && fc_status != BGWH_NOT_YET_STARTED)
		{
			/* 标记为已分离，以使其正式。 */
			fc_mq->mq_detached = true;
			return true;
		}
	}

	/* 对方并不确定已经消失。 */
	return false;
}

/*
 * 当一个进程在等待其对方附加到队列时使用。当另一个进程按预期附加时我们将退出，或者如果 handle != NULL，当引用的后台进程或进程主管死亡时。在 handle == NULL 的情况下，如果进程未能附加，我们可能会在这里永远卡住，等待一个可能永远不会启动的进程。然而，我们确实检查中断。
 *
 * ptr 是一个指向我们期待在对方附加到队列时变为非 NULL 的内存地址的指针。
 */
static bool fc_shm_mq_wait_internal(shm_mq *fc_mq, PGPROC **fc_ptr, BackgroundWorkerHandle *fc_handle)
{
	bool		fc_result = false;

	for (;;)
	{
		BgwHandleStatus fc_status;
		pid_t		fc_pid;

		/* 仅在检查指针时获取锁。 */
		SpinLockAcquire(&fc_mq->mq_mutex);
		fc_result = (*fc_ptr != NULL);
		SpinLockRelease(&fc_mq->mq_mutex);

		/* 如果已分离，则失败；否则如果已初始化则成功。 */
		if (fc_mq->mq_detached)
		{
			fc_result = false;
			break;
		}
		if (fc_result)
			break;

		if (fc_handle != NULL)
		{
			/* 检查意外的工作者死亡。 */
			fc_status = GetBackgroundWorkerPid(fc_handle, &fc_pid);
			if (fc_status != BGWH_STARTED && fc_status != BGWH_NOT_YET_STARTED)
			{
				fc_result = false;
				break;
			}
		}

		/* 等待被信号通知。 */
		(void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
						 WAIT_EVENT_MQ_INTERNAL);

		/* 重置锁以避免自旋。 */
		ResetLatch(MyLatch);

		/* 在我们等待时可能发生了中断。 */
		CHECK_FOR_INTERRUPTS();
	}

	return fc_result;
}

/*
 * 增加已读取的字节数。
 */
static void fc_shm_mq_inc_bytes_read(shm_mq *fc_mq, Size fc_n)
{
	PGPROC	   *fc_sender;

	/*
	 * 将先前读取的 mq_ring 与随后的 mq_bytes_read 增量分开。与 shm_mq_send_bytes() 中的完整屏障配对。我们在这里只需要一个读取屏障，因为 mq_bytes_read 的增量实际上是一次读取后跟一次依赖写入。
	 */
	pg_read_barrier();

	/*
	 * 在这里没有必要使用 pg_atomic_fetch_add_u64，因为没有其他人会更改这个值。此方法应该更便宜。
	 */
	pg_atomic_write_u64(&fc_mq->mq_bytes_read,
						pg_atomic_read_u64(&fc_mq->mq_bytes_read) + fc_n);

	/*
	 * 我们不应该在没有发送者的情况下有任何字节可读，因此我们可以在这里不加锁地读取 mq_sender。一旦初始化，它就无法更改。
	 */
	fc_sender = fc_mq->mq_sender;
	Assert(fc_sender != NULL);
	SetLatch(&fc_sender->procLatch);
}

/*
 * 增加已写入的字节数。
 */
static void fc_shm_mq_inc_bytes_written(shm_mq *fc_mq, Size fc_n)
{
	/*
	 * 将先前读取的 mq_ring 与即将进行的 mq_bytes_written 写入分开。与 shm_mq_receive_bytes 中找到的读取屏障配对。
	 */
	pg_write_barrier();

	/*
	 * 在这里没有必要使用 pg_atomic_fetch_add_u64，因为没有其他人会更改这个值。此方法避免了不必要地获取总线锁。
	 */
	pg_atomic_write_u64(&fc_mq->mq_bytes_written,
						pg_atomic_read_u64(&fc_mq->mq_bytes_written) + fc_n);
}

/* on_dsm_detach 回调的适配器。 */
static void fc_shm_mq_detach_callback(dsm_segment *fc_seg, Datum fc_arg)
{
	shm_mq	   *fc_mq = (shm_mq *) DatumGetPointer(fc_arg);

	fc_shm_mq_detach_internal(fc_mq);
}
